package qq2564874169.tuaotuao.fx.mq;


import qq2564874169.tuaotuao.fx.EventDispatcher;
import qq2564874169.tuaotuao.fx.FxStream;
import qq2564874169.tuaotuao.fx.Task;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;

public class AppMqSubscriber implements MqSubscriber {
    private final static String receivedKey = UUID.randomUUID().toString();
    private final static String errorKey = UUID.randomUUID().toString();
    /**
     * 本地缓存数量
     */
    public int prefetchCount = 200;
    private EventDispatcher<MqMessage> received = new EventDispatcher<>(receivedKey);
    private EventDispatcher<SubscriberError> error = new EventDispatcher<>(errorKey);
    private String eid;
    private Map<String, LinkedBlockingQueue<MsgContext>> msgQueue = new HashMap<>(8);
    private boolean isClosed;
    private Map<String, ConsumeContext> contextMap = new HashMap<>(8);

    @Override
    public void declare(String queue) {

    }

    @Override
    public EventDispatcher<MqMessage> evReceived() {
        return received;
    }

    @Override
    public EventDispatcher<SubscriberError> evError() {
        return error;
    }

    public AppMqSubscriber() {
        eid = MsgContext.evMsg.add(this::onReceived);
    }

    private void onReceived(Object sender, MsgContext message) {
        for (String k : new ArrayList<>(msgQueue.keySet())) {
            if (k.startsWith(message.queue + "$")) {
                LinkedBlockingQueue<MsgContext> queue = msgQueue.get(k);
                if (queue != null) {
                    try {
                        queue.put(message);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }
    }

    protected void onReceived(MqMessage message) {
        if (received != null) {
            received.fire(receivedKey, this, message);
        }
    }

    private AppMqMessage fireReceived(String queue, String name, String data, Object state, int count) {
        AppMqMessage msg = new AppMqMessage();
        msg.queue = queue;
        msg.name = name;
        msg.data = data;
        msg.state = state;
        msg.recount = count;
        onReceived(msg);
        if (msg.reconsume) {
            msg.recount++;
        }
        return msg;
    }

    @Override
    public String consume(String queue, String name, Object state) {
        ConsumeContext ctx = new ConsumeContext();
        ctx.route = getRoute(queue, name);
        ctx.name = name;
        ctx.queue = queue;
        ctx.state = state;
        contextMap.put(ctx.id, ctx);
        if (msgQueue.containsKey(ctx.route) == false) {
            msgQueue.put(ctx.route, new LinkedBlockingQueue<>(prefetchCount));
        }
        Task.run(s -> {
            ConsumeContext scc = (ConsumeContext) s;
            Queue<MsgContext> ctxQueue = msgQueue.get(scc.route);
            Queue<AppMqMessage> reMsgQueue = new LinkedBlockingQueue<>();
            while (msgQueue.containsKey(scc.route) && !isClosed && !scc.isStop) {
                try {
                    MsgContext item = ctxQueue.poll();
                    if (item == null) {
                        AppMqMessage reMsg = reMsgQueue.poll();
                        if (reMsg != null) {
                            AppMqMessage msg = fireReceived(scc.queue, scc.name, reMsg.data, scc.state, reMsg.recount);
                            if (msg.reconsume && msg.recount < 5) {
                                reMsgQueue.add(msg);
                            }
                        } else {
                            Task.delay(100, 200);
                        }
                    } else if (item.data != null && item.queue != null) {
                        AppMqMessage msg = fireReceived(scc.queue, scc.name, item.data, scc.state, 0);
                        if (msg.reconsume) {
                            reMsgQueue.add(msg);
                        }
                    }
                } catch (Exception e) {
                    SubscriberError args = new SubscriberError(e, state);
                    error.fire(errorKey, this, args);
                    if (args.isRetry == false) {
                        throw e;
                    }
                }
            }
            reMsgQueue.clear();
            ctxQueue.clear();
        }, ctx);
        return ctx.id;
    }

    @Override
    public void cancel(String queue, String name) {
        var mq = msgQueue.remove(getRoute(queue, name));
        if (mq != null) {
            mq.clear();
        }
        FxStream.create(contextMap.values())
                .filter(i -> i.queue.equals(queue) && i.name.equals(name))
                .foreach(i -> contextMap.remove(i.id).isStop = true);
    }

    @Override
    public void cancel(String consumeId) {
        var ctx = contextMap.remove(consumeId);
        if (ctx != null) {
            ctx.isStop = true;
            if (FxStream.create(contextMap.values()).all(i -> !i.route.equals(ctx.route))) {
                var mq = msgQueue.remove(ctx.route);
                if (mq != null) {
                    mq.clear();
                }
            }
        }
    }

    @Override
    public void close() {
        isClosed = true;
        msgQueue.clear();
        contextMap.clear();
        MsgContext.evMsg.remove(eid);
        if (received != null) {
            received.clear();
        }
    }

    private static String getRoute(String queue, String name) {
        return queue + "$" + name;
    }

    private static class AppMqMessage extends MqMessage {
        int recount;
    }

    private static class ConsumeContext {
        public String route;
        public Object state;
        public String queue;
        public String name;
        public final String id;
        public boolean isStop;

        public ConsumeContext() {
            id = UUID.randomUUID().toString();
        }
    }
}
